# coding: utf-8
# File   :  DingTalkBot
# Author :  wangsheng
# @Time  :  2025/5/9 上午10:09
import base64
import hashlib
import hmac
import time
import urllib

import requests
import json
import os
from typing import Optional, List, Dict, Union


class DingTalkBot:
    """
    钉钉群机器人通知工具类
    文档：https://open.dingtalk.com/document/orgapp/custom-robot-access
    """

    def __init__(self, webhook: str, secret: Optional[str] = None):
        """
        :param webhook: 机器人Webhook地址
        :param secret: 加签密钥（可选）
        """
        self.webhook = webhook
        self.secret = secret
        self.headers = {'Content-Type': 'application/json;charset=utf-8'}

    def _sign(self) -> Dict[str, str]:
        """生成加签时间戳（如果启用加签）"""
        if not self.secret:
            return {}

        timestamp = str(round(time.time() * 1000))
        secret_enc = self.secret.encode('utf-8')
        string_to_sign = f"{timestamp}\n{self.secret}"
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        return {
            "timestamp": timestamp,
            "sign": sign
        }

    def send_text(
            self,
            content: str,
            at_mobiles: Optional[List[str]] = None,
            at_all: bool = False
    ) -> bool:
        """
        发送文本消息
        :param content: 文本内容
        :param at_mobiles: 被@人的手机号列表
        :param at_all: 是否@所有人
        """
        data = {
            "msgtype": "text",
            "text": {"content": content},
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": at_all
            }
        }
        return self._send_request(data)

    def send_markdown(
            self,
            title: str,
            text: str,
            at_mobiles: Optional[List[str]] = None,
            at_all: bool = False
    ) -> bool:
        """
        发送Markdown消息
        :param title: 标题（会显示在通知列表）
        :param text: Markdown格式内容
        :param at_mobiles: 被@人的手机号列表
        :param at_all: 是否@所有人
        """
        data = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            },
            "at": {
                "atMobiles": at_mobiles or [],
                "isAtAll": at_all
            }
        }
        return self._send_request(data)

    def send_link(
            self,
            title: str,
            text: str,
            message_url: str,
            pic_url: Optional[str] = None
    ) -> bool:
        """
        发送链接消息
        :param title: 标题
        :param text: 内容描述
        :param message_url: 跳转链接
        :param pic_url: 图片URL（可选）
        """
        data = {
            "msgtype": "link",
            "link": {
                "title": title,
                "text": text,
                "messageUrl": message_url,
                "picUrl": pic_url or ""
            }
        }
        return self._send_request(data)

    def upload_file(self, file_path: str) -> Optional[str]:
        """
        上传文件到钉钉服务器（用于后续发送）
        :param file_path: 本地文件路径
        :return: 返回media_id（用于发送文件消息）
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件不存在: {file_path}")

        upload_url = "https://oapi.dingtalk.com/robot/upload?access_token=" + self.webhook.split("access_token=")[1]

        with open(file_path, 'rb') as f:
            files = {'media': f}
            response = requests.post(upload_url, files=files)

        if response.status_code == 200:
            return response.json().get('media_id')
        return None

    def send_file(self, media_id: str) -> bool:
        """
        发送文件消息
        :param media_id: 通过upload_file获取的media_id
        """
        data = {
            "msgtype": "file",
            "file": {
                "media_id": media_id
            }
        }
        return self._send_request(data)

    def _send_request(self, data: Dict) -> bool:
        """发送请求到钉钉API"""
        params = self._sign()
        try:
            response = requests.post(
                url=self.webhook,
                headers=self.headers,
                data=json.dumps(data),
                params=params,
                timeout=10
            )
            result = response.json()
            if result.get('errcode') != 0:
                print(f"钉钉机器人发送失败: {result.get('errmsg')}")
                return False
            return True
        except Exception as e:
            print(f"钉钉机器人请求异常: {str(e)}")
            return False


# ==================== 使用示例 ====================
if __name__ == "__main__":
    # 初始化机器人（需替换为你的webhook和secret）
    bot = DingTalkBot(
        webhook="https://oapi.dingtalk.com/robot/send?access_token=1ea5f0143d7e6f6973b65097427dd9aa1783bdecbe38f5e5a4b56689c9b274dc",
        secret="SEC5abf73b8862b7705934b048a59f5560e79ad456d392ee017901c80b0a4807aae"  # 如果没启用加签可以去掉
    )

    # 示例1：发送文本消息（@指定人）
    bot.send_text(
        content="测试环境部署完成\n请及时验证",
        at_mobiles=["13800138000"]
    )

    # 示例2：发送Markdown消息（适合测试报告）
    bot.send_markdown(
        title="自动化测试报告",
        text="""### 测试结果
- ​**总共用例**: 42
- ​**通过用例**: 42
- ​**失败用例**: 3
- ​**成功率**: 93.3%
- ​**耗时**: xxx
- ​**报告链接**: [点击查看](http://your-report-url)""",
        at_all=True
    )